package com.atguigu.flink.chapter05.sink;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.Arrays;
import java.util.Properties;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/22 10:14
 */
public class Flink02_Redis {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
//                .fromElements(
//                        new WaterSensor("sensor_1", 1L, 1),
//                        new WaterSensor("sensor_2", 2L, 2),
//                        new WaterSensor("sensor_3", 3L, 3)
//
//                );
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");
                        return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
                    }
                });

        // TODO Sink Redis
        FlinkJedisPoolConfig jedisConfig = new FlinkJedisPoolConfig.Builder()
                .setHost("hadoop102")
                .setPort(6379)
                .build();

        RedisSink<WaterSensor> redisSinkFunction = new RedisSink<>(jedisConfig,
                new RedisMapper<WaterSensor>() {
                    @Override
                    public RedisCommandDescription getCommandDescription() {
                        // 第二个参数是redis最外层的key
                        //    key               value(Hash)
                        //    flink0820         field=value
                        return new RedisCommandDescription(RedisCommand.HSET, "flink0820");
                    }

                    @Override
                    public String getKeyFromData(WaterSensor s) {
                        // 从数据里提取key，如果是hash结构，提取的就是 field（hash结构的key）
                        return s.getId();
                    }

                    @Override
                    public String getValueFromData(WaterSensor s) {
                        // 从数据里提取value，如果是hash结构，提取的就是 Hash的value
                        return s.getVc().toString();
                    }
                });

        sensorDS.addSink(redisSinkFunction);

        env.execute();
    }
}
